package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.TradeOrderBean;
import com.atguigu.edu.realtime.util.ClickHouseUtil;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * @author Lec
 * @date 2022/9/5 15:21
 */

public class DwsTradeOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 2. 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        */


        // TODO 3. 从 Kafka dwd_trade_order_detail 主题读取下单明细数据
        String topicName = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_window";
        DataStreamSource<String> orderStreamSource = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));

//        orderStreamSource.print();

        // TODO 4 转换结构为TradeOrderBean
        SingleOutputStreamOperator<TradeOrderBean> beanStream = orderStreamSource.map(new MapFunction<String, TradeOrderBean>() {
            @Override
            public TradeOrderBean map(String value) throws Exception {
                HashSet<String> orderSet = new HashSet<>();
                JSONObject jsonObject = JSONObject.parseObject(value);
                String sc = jsonObject.getString("sc");
                String provinceId = jsonObject.getString("province_id");
                String courseId = jsonObject.getString("course_id");
                String userId = jsonObject.getString("user_id");
                String orderId = jsonObject.getString("order_id");
                Double originAmount = jsonObject.getDouble("origin_amount");
                Long ts = jsonObject.getLong("od_ts");
                orderSet.add(orderId);
                return TradeOrderBean.builder()
                        .sc(sc)
                        .provinceId(provinceId)
                        .courseId(courseId)
                        .orderIdSet(orderSet)
                        .userId(userId)
                        .orderAmount(originAmount)
                        .ts(ts * 1000L)
                        .build();
            }
        });
//        beanStream.print("beanstream---------->");
        //TODO 5. 添加水位线
        SingleOutputStreamOperator<TradeOrderBean> watermarkStream = beanStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TradeOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                .withTimestampAssigner(new SerializableTimestampAssigner<TradeOrderBean>() {
                    @Override
                    public long extractTimestamp(TradeOrderBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        //TODO 6. 开窗聚合
        KeyedStream<TradeOrderBean, String> keyedStream = watermarkStream.keyBy(new KeySelector<TradeOrderBean, String>() {
            @Override
            public String getKey(TradeOrderBean value) throws Exception {
                String categoryId = value.getCategoryId();
                String userId = value.getUserId();
                String provinceId = value.getProvinceId();
                String sc = value.getSc();
                String key = categoryId + "_" + userId + "_" + provinceId + "_" + sc;
                return key;
            }
        });

        WindowedStream<TradeOrderBean, String, TimeWindow> windowedStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10L)));

        SingleOutputStreamOperator<TradeOrderBean> reduceStream = windowedStream.reduce(new ReduceFunction<TradeOrderBean>() {
            @Override
            public TradeOrderBean reduce(TradeOrderBean value1, TradeOrderBean value2) throws Exception {
                value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                Set<String> o2Set = value2.getOrderIdSet();
                Set<String> o1Set = value1.getOrderIdSet();
                o1Set.addAll(o2Set);
//                value1.setOrderIdSet(value1.getOrderIdSet().add(value2.getOrderIdSet().toString()));
                value1.setOrderIdSet(o1Set);

                return value1;
            }
        }, new WindowFunction<TradeOrderBean, TradeOrderBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<TradeOrderBean> input, Collector<TradeOrderBean> out) throws Exception {

                    TradeOrderBean tradeOrderBean = input.iterator().next();
                    tradeOrderBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                    tradeOrderBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                    int size = tradeOrderBean.getOrderIdSet().size();
                    tradeOrderBean.setOrderCount((long) size);
                    tradeOrderBean.setUserCount(1L);
                    tradeOrderBean.setTs(System.currentTimeMillis());
                    out.collect(tradeOrderBean);


            }
        });

        // TODO 7 . 补充其他字段
        SingleOutputStreamOperator<TradeOrderBean> courseBean = AsyncDataStream.unorderedWait(reduceStream, new DimAsyncFunction<TradeOrderBean>("DIM_COURSE_INFO") {
            @Override
            public String getKey(TradeOrderBean input) {
                return input.getCourseId();
            }

            @Override
            public void join(TradeOrderBean input, JSONObject obj) {
                input.setCourseName(obj.getString("courseName"));
                input.setSubjectId(obj.getString("subjectId"));
            }
        }, 100L, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeOrderBean> subjectBean = AsyncDataStream.unorderedWait(courseBean, new DimAsyncFunction<TradeOrderBean>("DIM_BASE_SUBJECT_INFO") {
            @Override
            public String getKey(TradeOrderBean input) {
                return input.getSubjectId();
            }

            @Override
            public void join(TradeOrderBean input, JSONObject obj) {
                input.setSubjectName(obj.getString("subjectName"));
                input.setCategoryId(obj.getString("categoryId"));
            }
        }, 100L, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeOrderBean> categoryBean = AsyncDataStream.unorderedWait(subjectBean, new DimAsyncFunction<TradeOrderBean>("DIM_BASE_CATEGORY_INFO") {
            @Override
            public String getKey(TradeOrderBean input) {
                return input.getCategoryId();
            }

            @Override
            public void join(TradeOrderBean input, JSONObject obj) {
                input.setCategoryName(obj.getString("categoryName"));
            }
        }, 100L, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeOrderBean> provinceBean = AsyncDataStream.unorderedWait(categoryBean, new DimAsyncFunction<TradeOrderBean>("DIM_BASE_PROVINCE") {
            @Override
            public String getKey(TradeOrderBean input) {
                return input.getProvinceId();
            }

            @Override
            public void join(TradeOrderBean input, JSONObject obj) {
                input.setProvinceName(obj.getString("name"));
            }
        }, 100L, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeOrderBean> result = AsyncDataStream.unorderedWait(provinceBean, new DimAsyncFunction<TradeOrderBean>("DIM_BASE_SOURCE") {
            @Override
            public String getKey(TradeOrderBean input) {
                return input.getSc();
            }

            @Override
            public void join(TradeOrderBean input, JSONObject obj) {
                input.setScName(obj.getString("sourceSite"));
            }
        }, 100L, TimeUnit.SECONDS);

        result.print();

        //TODO 8.写入到clickhouse中
        result.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_trade_order_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

        // TODO 9.执行任务
        env.execute();

    }
}
